package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //1、接收信息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //2、聚合流式处理
        stream.map((key, value) -> {
                    //3、key是值 具体消息序列化成成json存储在value中
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    //4、重新分配key 以及 value
                    return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name()+ ":" + mess.getAdd());
                })
                //5、按照文章id分组聚合
                .groupBy((key, value) -> key)
                //6、创建时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //7、自行完成聚合的计算（不是普通的count就可以实现的，这里有第三个参数Materialized.as(String string) 表示stream流的名称 任意，不重复即可）
                .aggregate(new Initializer<String>() {
                    //7.1、初始化方法，每个时间窗口内进行聚合前都会执行一次，且只执行一次 返回值时消息的value
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    //7.2、对stream中的每个消息进行聚合处理时都会执行一次该方法 真正的聚合操作，返回值时消息的value
                    @Override
                    public String apply(String key, String value, String aggValue) {
                        //7.3、非空校验
                        if (StringUtils.isBlank(value)) {
                            return aggValue;
                        }
                        //切割初始化值
                        String[] aggAry = aggValue.split(",");
                        // 定义初始化值
                        int col = 0, com = 0, lik = 0, vie = 0;
                        // 遍历 得到每个行为及其分数
                        for (String agg : aggAry) {
                            // 切割 将每个行为及其对应数值分开
                            String[] split = agg.split(":");
                            /**
                             * 获得初始值，也就是时间窗口内计算之后的值
                             */
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                                case LIKES:
                                    lik = Integer.parseInt(split[1]);
                                    break;
                                case COLLECTION:
                                    col = Integer.parseInt(split[1]);
                                    break;
                                case VIEWS:
                                    vie = Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(split[1]);
                                    break;
                            }
                        }
                        /**
                         * 累加操作
                         */
                        //切割value
                        String[] valAry = value.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])) {
                            case LIKES:
                                lik += Integer.parseInt(valAry[1]);
                                break;
                            case COLLECTION:
                                col += Integer.parseInt(valAry[1]);
                                break;
                            case COMMENT:
                                com += Integer.parseInt(valAry[1]);
                                break;
                            case VIEWS:
                                vie += Integer.parseInt(valAry[1]);
                                break;
                        }
                        //格式化结果数据
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        log.info("文章id：{}", key);
                        log.info("当前时间窗口内的消息处理结果：{}", formatStr);
                        //返回最后将结果
                        return formatStr;
                    }
                })
                //8、转换成kstream
                .toStream()
                //9、算分后再次重新分配key value 需要知道文章id
                .map((key, value) -> {
                    return new KeyValue<>(key, formatObj(key.key().toString(), value.toString()));
                })
                //10、往消费者发送消息
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        //11、返回kstram对象
        return stream;
    }

    /**
     * 格式化消息的value数据
     *
     * @param articleId
     * @param value
     * @return
     */
    private String formatObj(String articleId, String value) {
        //1、创建需返回对象
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        //2、完善属性
        //2.1、键为文章id
        mess.setArticleId(Long.valueOf(articleId));
        //2.2、切割value得到每个行为以及对应的数值 COLLECTION:1 LIKES:0
        String[] valAry = value.split(",");
        //2.3、遍历数组
        for (String val : valAry) {
            //2.4、再次切割:  COLLECTION 1
            String[] split = val.split(":");
            //2.5、switch判断 第一位 即零号索引 jdk5开始引入枚举 7开始引入String
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理之后的结果为：{}", JSON.toJSONString(mess));
        return JSON.toJSONString(mess);
    }
}
